package com.david.kafka.auth.handler;


import cn.hutool.core.collection.CollUtil;
import com.alibaba.druid.pool.DruidDataSource;
import com.david.kafka.auth.cache.LocalCache;
import com.david.kafka.auth.constants.Constants;
import com.david.kafka.auth.bean.RoleResourceResultVO;
import com.david.kafka.auth.bean.RoleUserResultVO;
import com.david.kafka.auth.util.DataSourceUtil;
import org.apache.kafka.common.Endpoint;
import org.apache.kafka.common.acl.AclBinding;
import org.apache.kafka.common.acl.AclBindingFilter;
import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.resource.PatternType;
import org.apache.kafka.common.resource.ResourcePattern;
import org.apache.kafka.common.resource.ResourceType;
import org.apache.kafka.server.authorizer.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * db acl鉴权类
 * 配置方法：在server.properties添加如下配置：
 * super.users 超级用户，多个用;隔开
 * authorizer.class.name=com.openvalley.kafka.auth.handler.DbAclAuthorizer
 *
 * @author David
 */
public class DbAclAuthorizer implements Authorizer {

    private static final Logger LOG = LoggerFactory.getLogger(DbAclAuthorizer.class);

    private static final String SUPER_USERS_PROP = "super.users";
    private DruidDataSource dataSource = null;
    private Set<String> superUserSet;

    /**
     * resourceSystemId
     */
    private String resourceSystemId;
    /**
     * tenantId
     */
    private String tenantId;

    private static final Logger log = LoggerFactory.getLogger(DbAclAuthorizer.class.getName());

    @Override
    public Map<Endpoint, ? extends CompletionStage<Void>> start(AuthorizerServerInfo serverInfo) {
        log.info("start");
        return new HashMap<>();
    }

    @Override
    public List<AuthorizationResult> authorize(AuthorizableRequestContext requestContext, List<Action> actions) {
        return actions.stream().map(action -> authorizeAction(requestContext, action)).collect(Collectors.toList());
    }

    private AuthorizationResult authorizeAction(AuthorizableRequestContext requestContext, Action action) {
        ResourcePattern resource = action.resourcePattern();
        if (resource.patternType() != PatternType.LITERAL) {
            throw new IllegalArgumentException("Only literal resources are supported. Got: " + resource.patternType());
        }
        String principal = requestContext.principal().getName();

        AclOperation operation = action.operation();
        //1 超级用户直接通过
        if (superUserSet.contains(principal)) {
            return AuthorizationResult.ALLOWED;
        }
        //2 资源类型为 Cluster 直接不通过
        if (resource.resourceType().equals(ResourceType.CLUSTER)) {
            return AuthorizationResult.DENIED;
        }
        //3 资源类型为 TransactionalId、DelegationToken 直接通过
        if (resource.resourceType().equals(ResourceType.TRANSACTIONAL_ID) || resource.resourceType().equals(ResourceType.DELEGATION_TOKEN)) {
            return AuthorizationResult.ALLOWED;
        }
        //4 寻找资源对应的权限配置信息，找到则通过，否则不通过
        if (matchingAcls(resource.resourceType(), resource.name(), operation, principal)) {
            return AuthorizationResult.ALLOWED;
        }
        return AuthorizationResult.DENIED;
    }

    /**
     * @param resourceType
     * @param resourceName
     * @param operation
     * @param principal
     * @return
     */
    private Boolean matchingAcls(ResourceType resourceType, String resourceName, AclOperation operation, String principal) {
        List<RoleUserResultVO> voList = LocalCache.getCache(principal);
        if (CollUtil.isEmpty(voList)) {
            List<RoleUserResultVO> voListDb = new ArrayList<>();
            //从数据库查询
            String userId = getUserId(principal);
            if (StringUtils.hasText(userId)) {
                return Boolean.FALSE;
            }
            setRoleUser(userId, voListDb);
            setRoleResourceUser(voListDb);
            LocalCache.addCache(principal, voListDb);
            Boolean b = authTopic(resourceName, voListDb);
            log.info("从数据库中验证topic:{}权限,用户{},结果{}", resourceName, principal, b);
            return b;

        }

        Boolean b = authTopic(resourceName, voList);
        log.info("从缓存中验证topic:{}权限,用户{},结果{}", resourceName, principal, b);
        return b;
    }


    private Boolean authTopic(String resourceName, List<RoleUserResultVO> voList) {
        for (RoleUserResultVO resultVO : voList) {
            List<RoleResourceResultVO> roleResourceResultVOS = resultVO.getRoleResourceResultVOS();
            if (CollUtil.isEmpty(roleResourceResultVOS)) {
                continue;
            }
            if (topicRoleResourceMatch(resourceName, roleResourceResultVOS)) {
                return Boolean.TRUE;
            }
        }

        return Boolean.FALSE;
    }

    private String getUserId(String username) {
        String userQuery = "SELECT\n" +
                "        t1.id,\n" +
                "        t1.username,\n" +
                "        t1.password,\n" +
                "            itu.tenant_id tenantId\n" +
                "        FROM\n" +
                "            idaas_user t1\n" +
                "            LEFT JOIN idaas_tenant_user itu on t1.id = itu.user_id\n" +
                "       WHERE \n" +
                "            itu.tenant_id = ? \n" +
                "            and t1.del_flag = 0\n" +
                "            AND t1.username = ?\n" +
                "            AND t1.status ='ENABLE'\n" +
                "        LIMIT 1";
        try (Connection conn = dataSource.getConnection();
             PreparedStatement statement = conn.prepareStatement(userQuery);
        ) {
            statement.setString(1, tenantId);
            statement.setString(2, username);

            try (ResultSet resultSet = statement.executeQuery()) {
                if (resultSet.next()) {
                    String userId = resultSet.getString("id");
                    return userId;
                } else {
                    LOG.info("鉴权失败,user {} not found in db.", username);
                    return "";
                }
            }
        } catch (SQLException e) {
            LOG.error("数据库查询用户异常{}", e);
            throw new RuntimeException(e);
        }
    }

    /**
     * 获取角色权限
     *
     * @param voListDb
     */
    private void setRoleResourceUser(List<RoleUserResultVO> voListDb) {
        for (RoleUserResultVO resultVO : voListDb) {
            List<RoleResourceResultVO> roleResourceResultVOS = new ArrayList<>();
            String roleQuery = "SELECT\n" +
                    "r.id,\n" +
                    "r.parent_id,\n" +
                    "r.resource_name,\n" +
                    "r.resource_type,\n" +
                    "r.path_url\n" +
                    "FROM\n" +
                    "idaas_resource r\n" +
                    "LEFT JOIN idaas_role_resource rr ON ( rr.resource_id = r.id AND rr.role_id = ? )\n" +
                    "WHERE\n" +
                    " r.resource_system_id = ? \n" +
                    "or r.resource_system_id = (SELECT raw_sys_id FROM idaas_resource_system WHERE id = ? )\n" +
                    "AND r.`status` = 'ENABLE'\n" +
                    "AND r.del_flag = FALSE";
            try (Connection conn = dataSource.getConnection();
                 PreparedStatement statement = conn.prepareStatement(roleQuery);
            ) {

                statement.setString(1, resultVO.getRole_id());
                statement.setString(2, resourceSystemId);
                statement.setString(3, resourceSystemId);
                try (ResultSet resultSet = statement.executeQuery()) {
                    while (resultSet.next()) {
                        RoleResourceResultVO resourceResultVO = new RoleResourceResultVO();
                        resourceResultVO.setResource_name(resultSet.getString("resource_name"));
                        resourceResultVO.setPath_url(resultSet.getString("path_url"));
                        resourceResultVO.setId(resultSet.getString("id"));
                        resourceResultVO.setParent_id(resultSet.getString("parent_id"));
                        roleResourceResultVOS.add(resourceResultVO);
                    }
                }
            } catch (SQLException e) {
                LOG.error("查询idaas数据库,获取角色权限信息报错{}");
                throw new RuntimeException(e);
            }
            resultVO.setRoleResourceResultVOS(roleResourceResultVOS);
        }

    }


    /**
     * 获取角色用户
     *
     * @param userId
     * @param voListDb
     */
    private void setRoleUser(String userId, List<RoleUserResultVO> voListDb) {
        String roleQuery = "SELECT\n" +
                "roleUser.user_id,\n" +
                "roleUser.role_id\n" +
                "FROM\n" +
                "idaas_role_user AS roleUser\n" +
                "LEFT JOIN idaas_role rol ON rol.id = roleUser.role_id\n" +
                "LEFT JOIN idaas_user AS u ON roleUser.user_id = u.id\n" +
                "WHERE  \n" +
                "    rol.resource_system_id = ?\n" +
                "AND roleUser.user_id = ?\n";
        try (Connection conn = dataSource.getConnection();
             PreparedStatement statement = conn.prepareStatement(roleQuery);
        ) {

            statement.setString(1, resourceSystemId);
            statement.setString(2, userId);
            try (ResultSet resultSet = statement.executeQuery()) {
                while (resultSet.next()) {
                    RoleUserResultVO resultVO = new RoleUserResultVO();
                    resultVO.setRole_id(resultSet.getString("role_id"));
                    resultVO.setUser_id(resultSet.getString("user_id"));
                    voListDb.add(resultVO);
                }
            }
        } catch (SQLException e) {
            LOG.error("查询idaas数据库,获取角色信息报错{}");
            throw new RuntimeException(e);
        }
    }

    /**
     * 角色资源中是否配置了此topic的权限
     *
     * @param topic
     * @param resultVOList
     * @return
     */
    private Boolean topicRoleResourceMatch(String topic, List<RoleResourceResultVO> resultVOList) {
        for (RoleResourceResultVO resultVO : resultVOList) {
            if (StringUtils.isEmpty(resultVO.getPath_url())) {
                continue;
            }

            if (isMatch(topic, resultVO.getPath_url())) {
                return Boolean.TRUE;
            }
        }
        return Boolean.FALSE;
    }

    /**
     * 创建权限只能通过idaas
     *
     * @param requestContext
     * @param aclBindings
     * @return
     */
    @Override
    public List<? extends CompletionStage<AclCreateResult>> createAcls(AuthorizableRequestContext requestContext, List<AclBinding> aclBindings) {
        LOG.error("只能通过idaas进行权限操作!");
        throw new UnsupportedOperationException();
    }

    /**
     * 删除权限只能通过idaas
     *
     * @param requestContext
     * @param aclBindingFilters
     * @return
     */
    @Override
    public List<? extends CompletionStage<AclDeleteResult>> deleteAcls(AuthorizableRequestContext requestContext, List<AclBindingFilter> aclBindingFilters) {
        LOG.error("只能通过idaas进行权限操作!");
        throw new UnsupportedOperationException();
    }

    @Override
    public Iterable<AclBinding> acls(AclBindingFilter filter) {
        LOG.error("只能通过idaas进行权限操作!");
        throw new UnsupportedOperationException();
    }

    @Override
    public void close() {
        log.info("close");
        if (dataSource != null) {
            dataSource.close();
        }
    }

    @Override
    public void configure(Map<String, ?> configs) {
        log.info("configure");
        String superUsers = (String) configs.get(SUPER_USERS_PROP);
        log.info("superUsers:{}", superUsers);
        if (superUsers == null || superUsers.isEmpty()) {
            superUserSet = new HashSet<>();
        } else {
            superUserSet = Arrays.stream(superUsers.split(";")).map(String::trim).collect(Collectors.toSet());
        }

        resourceSystemId = (String) configs.get(Constants.IDAAS_RESOURCE_SYSTEMID);

        dataSource = DataSourceUtil.getInstance(configs);
        tenantId = (String) configs.get(Constants.IDAAS_TENANT_ID);
    }

    /**
     * 匹配字符串  单个匹配? # 多个匹配*
     *
     * @param s    被匹配的字符串
     * @param rule 匹配规则
     * @return 匹配成功 返回true 失败返回 false
     */
    private static boolean isMatch(final String s, final String rule) {
        int len1 = s.length();
        int len2 = rule.length();
        boolean[][] dp = new boolean[len1 + 1][len2 + 1];
        dp[len1][len2] = true;
        for (int i = len1; i >= 0; i--) {
            for (int j = len2; j >= 0; j--) {
                if (i == len1 && j == len2) {
                    continue;
                }
                boolean firstMatch = (i < len1 && j < len2 && (s.charAt(i) == rule.charAt(j) || rule.charAt(j) == '?' || rule.charAt(j) == '#' || rule.charAt(j) == '*'));
                if (j < len2 && rule.charAt(j) == '*') {
                    dp[i][j] = dp[i][j + 1] || firstMatch && dp[i + 1][j];
                } else {
                    dp[i][j] = firstMatch && dp[i + 1][j + 1];
                }
            }
        }
        return dp[0][0];
    }
}
